package com.surfilter.massdata.spark.output;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.text.SimpleDateFormat;
import java.util.Date;
//import java.util.Iterator;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.collection.Iterator;

import com.act.sparkanalyz.output.impl.HiveOutput;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.JdbcUtil;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.iPCheckZjUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpUtils;
import com.surfilter.massdata.spark.util.ipcheck.JavaForeachPartitionDFFunc;

public class IpCheckZjStartBr2002 extends HiveOutput {

	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(IpCheckZjStartBr2002.class);
	private String table;
	private String columns;
	private String values;
	private String telecomIspid = null;
	private String mobileIspid = null;
	private String unicomIspid = null;
	private String railcomIspid = null;
	
	//private static JavaSparkContext ctx;
	
	public void write(DataFrame result, Map<String, String> commandMap) {
		
		log.info("===========IpCheckZjStartCleanBr2002 begin============");
		System.out.println("第三阶段--------");
		try{
			System.err.println("br2002 result>>>>>>>>>>>>>>>>>>>>>>>>"+result.count());
			/*在往BR2002表里插入数据之前，要truncate数据，在truncate数据之前要触发数据的结果将其让如内存，否则truncate之后数据在执行没有任何意义*/
			//result.count();   
			System.out.println("获取数据完成");
			IpCheckUtil.truncateTable(table);//清空表数据
			System.out.println("清空BR2002表");
			log.info("===========Clean Br2002 over============");
			writeToTable(result,table);
			log.info("===========Insert into Br2002 over============");
		}catch(Exception e){
			System.out.println("插入数据err:"+iPCheckZjUtil.getLineInfo());
			e.printStackTrace();
		}
		
		log.info("===========IpCheckZjStartCleanBr2002 end============");
	}

	private void writeToTable(DataFrame result, String table) {
		
		
		System.out.println("读取运营商配置信息......");
		
		telecomIspid = ConfigUtil.get("surfilter.ipcheck.telecom.ispid");// 电信
		mobileIspid = ConfigUtil.get("surfilter.ipcheck.mobile.ispid");// 移动
		unicomIspid = ConfigUtil.get("surfilter.ipcheck.unicom.ispid");// 联通
		if(!"null".equals(ConfigUtil.get("surfilter.ipcheck.railcom.ispid"))){
			railcomIspid = ConfigUtil.get("surfilter.ipcheck.railcom.ispid");// 铁通
		}
		
		JavaSparkContext jsc = JavaSparkContext.fromSparkContext(result.sqlContext().sparkContext());
		final Broadcast<Map<Long, String>> broadcastBr3010Map=jsc.broadcast(iPCheckZjUtil.getBR3010Map());
		final Broadcast<Map<String, String>> broadcastSY0002Map=jsc.broadcast(iPCheckZjUtil.getSY0002Map());
		
		
		final String br2002InsertSql = "INSERT INTO "+table+columns+" "+values;
		
		System.out.println("开始入库......");
		
		result.foreachPartition(new JavaForeachPartitionDFFunc() {
		//result.toJavaRDD().foreachPartition(new VoidFunction<Iterator<Row>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<Row> it) {
				Connection conn = null;
				PreparedStatement br_finally = null;
				try{
					//TODO   
					conn = JdbcUtil.getConnection();
					conn.setAutoCommit(false);
					int count = 0;
					int batchSize = 5000;
					br_finally = conn.prepareStatement(br2002InsertSql);
					//System.out.println("开始向表中插入数据......");
					while(it.hasNext()){
						Row row = it.next();
						Long key = row.getAs("bbdw_id");
						String mapValue = broadcastBr3010Map.getValue().get(key);
						String ispName = "";
						String ispBh = "";
						if (mapValue != null) {
							String[] valueArr = mapValue.split(StringUtil.STR_DELIMIT_1ST);
							if (valueArr.length == 2) {
								ispName = valueArr[0];
								ispBh = valueArr[1];
							}
						}
						
						br_finally.setString(1,IpUtils.longToIP(Long.parseLong(row.getAs("qsip").toString())));// 1
						br_finally.setString(2,IpUtils.longToIP(Long.parseLong(row.getAs("zzip").toString())));// 2
						br_finally.setLong(3,Long.parseLong(row.getAs("qsip").toString()));// 3
						br_finally.setLong(4,Long.parseLong(row.getAs("zzip").toString()));// 4
						br_finally.setString(5,row.getAs("bbdw_id").toString());// 5
						if (row.getAs("syfs").equals(-1l)) {// 6
							br_finally.setNull(6,Types.VARCHAR);
						} else {
							br_finally.setString(6,row.getAs("syfs").toString());
						}
						// 7 8
						if (row.getAs("bbdw_id").equals(Long.parseLong(telecomIspid))) { // 电信
							br_finally.setString(7,"1");
							br_finally.setString(8,broadcastSY0002Map.getValue().get("1"));
						} else if (row.getAs("bbdw_id").equals(Long.parseLong(mobileIspid))) { // 移动
							br_finally.setString(7,"2");
							br_finally.setString(8,broadcastSY0002Map.getValue().get("2"));
						} else if (row.getAs("bbdw_id").equals(Long.parseLong(unicomIspid))) { // 联通
							br_finally.setString(7,"3");
							br_finally.setString(8,broadcastSY0002Map.getValue().get("3"));
						} else if (railcomIspid != null && row.getAs("bbdw_id").equals(Long.parseLong(railcomIspid))) {
							br_finally.setString(7,"5");
							if (broadcastSY0002Map.getValue().get("5") != null) {
								br_finally.setString(8,broadcastSY0002Map.getValue().get("5"));
							} else {
								br_finally.setString(8,"其他");
							}
						} else {
							br_finally.setString(7,"4");
							br_finally.setString(8,"其他");
						}
						br_finally.setNull(9,Types.VARCHAR);// 9
						br_finally.setString(10,ispName);// 10
						br_finally.setString(11,ispBh);// 11
						if (row.getAs("shengid").equals(-1l)
								|| (row.getAs("shengid").equals(0l))) {// 12
							br_finally.setNull(12,Types.VARCHAR);
						} else {
							br_finally.setString(12,(row.getAs("shengid").toString()));
						}
						if (row.getAs("shiid").equals(-1l)
								|| row.getAs("shiid").equals(0l)) {// 13
							br_finally.setNull(13,Types.VARCHAR);
						} else {
							br_finally.setString(13,row.getAs("shiid").toString());
						}
						if (row.getAs("xianid").equals(-1l)
								|| row.getAs("xianid").equals(0l)) {// 14
							br_finally.setNull(14,Types.VARCHAR);
						} else {
							br_finally.setString(14,row.getAs("xianid").toString());
						}
						if (row.getAs("xxdz").equals("N/A")) {// 15
							br_finally.setNull(15,Types.VARCHAR);
						} else {
							br_finally.setString(15,row.getAs("xxdz").toString());
						}
						if (row.getAs("houseCode").equals("N/A")) {// 16
							br_finally.setNull(16,Types.VARCHAR);
						} else {
							br_finally.setString(16,row.getAs("houseCode").toString());
						}
						if (row.getAs("houseName").equals("N/A")) {// 17
							br_finally.setNull(17,Types.VARCHAR);
						} else {
							br_finally.setString(17,row.getAs("houseName").toString());
						}
						if (row.getAs("fpfs").equals(-1L)) {// 18
							br_finally.setNull(18,Types.DATE);
						} else {
							br_finally.setString(18,row.getAs("fpfs").toString());
						}
						if (row.getAs("jtiplx").equals(-1)) {// 19
							br_finally.setNull(19,Types.VARCHAR);
						} else {
							br_finally.setString(19,row.getAs("jtiplx").toString());
						}
						br_finally.setString(20,new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));// 20
						if (row.getAs("dwmc").equals("N/A")) {// 21
							br_finally.setNull(21,Types.VARCHAR);
						} else {
							br_finally.setString(21,row.getAs("dwmc").toString());
						}
						if (row.getAs("lxrxm").equals("N/A")) {// 22
							br_finally.setNull(22,Types.VARCHAR);
						} else {
							br_finally.setString(22,row.getAs("lxrxm").toString());
						}
						if (row.getAs("lxrdh").equals("N/A")) {// 23
							br_finally.setNull(23,Types.VARCHAR);
						} else {
							br_finally.setString(23,row.getAs("lxrdh").toString());
						}
						//TODO
						if (row.getAs("dwxz").equals(-1l) || row.getAs("dwxz").equals(null)) {// 24
							//br_finally.setString(24,"0");
							br_finally.setNull(24,Types.VARCHAR);
						} else {
							br_finally.setString(24,row.getAs("dwxz").toString());
						}
						if (row.getAs("dwhyfl").equals(-1l)) {// 25
							br_finally.setNull(25,Types.VARCHAR);
						} else {
							br_finally.setString(25,row.getAs("dwhyfl").toString());
						}
						if (row.getAs("bz").equals("N/A")) {// 26
							br_finally.setNull(26,Types.VARCHAR);
						} else {
							br_finally.setString(26,row.getAs("bz").toString());
						}
						if (row.getAs("sourceIdcId").equals("N/A")) {// 27
							br_finally.setNull(27,Types.VARCHAR);
						} else {
							br_finally.setString(27,row.getAs("sourceIdcId").toString());
						}
						br_finally.addBatch();
						if (++count % batchSize == 0) {
							br_finally.executeBatch();
							conn.commit();
							br_finally.clearBatch();
							count = 0;
						}

					}//end while
					br_finally.executeBatch();
					conn.commit();
					br_finally.clearBatch();
					//System.out.println("数据入库完成  partition end ");
				}catch(Exception e){
					System.err.println("入库失败~");
					e.printStackTrace();
				}finally{
					if(null != br_finally){
						try{
							br_finally.close();
						}catch(Exception e){
							e.printStackTrace();
						}
					}
					if (conn != null) {
						try {
							conn.close();
						} catch (SQLException e) {
							e.printStackTrace();
						}
					}
				}
			}
		});
		System.out.println("数据入库完成  end");
	}

}




